﻿#include "AvType.h"
#include "AvProto.h"
#include "ProtoHeader-c.h"
#include "net.h"
#include "AvProtoMsg.h"
#include "Queue.h"
#include "log.h"


typedef	struct {
	QUEUE ptr;
	LogItem	Iteam;
}LogIteamQueue;

typedef struct {
	char				uid[64];
	char				uuid[64];
	char				MacAddr[128];
	int					bThreadRun;
	int					bNetConnected;

	CMsgHead* MsgHeadRecv;
	CMsgHead* MsgHeadSend;
	char* MsgBuffer;

	time_t				LastSendKeepAliveT;
	LogIteamQueue		LogList;
	int					LogListCount;
#if defined(_B_HAVE_SOCKETPAIR)
	int					SockPairIn;
	int					SockPairOut;
#else
	avMutexT			LogMutex;
#endif
	onServiceEvent		onEvent;

}AvProtoHandle;


static AvProtoHandle* gAvProtoHandle = NULL;

int _RequestConnect(char* data) {
	C_Connect _Connect;
	memset(&_Connect, 0x00, sizeof(C_Connect));
	if (0 < strlen(gAvProtoHandle->uid)) {
		sprintf(_Connect.UID, "%s", gAvProtoHandle->uid);
	}
	if (0 < strlen(gAvProtoHandle->uuid)) {
		sprintf(_Connect.UUID, "%s", gAvProtoHandle->uuid);
	}
	if (0 == strlen(gAvProtoHandle->MacAddr)) {
		C_NetCommProfile _NetCommProfile = { 0 };
		gAvProtoHandle->onEvent(_ukProtoGet | __MsgNetCommProfile, &_NetCommProfile);
		switch (_NetCommProfile.NetDevice)
		{
		case NetCommT_LAN:
			sprintf(gAvProtoHandle->MacAddr, "eth0@%s", _NetCommProfile.Mac);
			break;
		case NetCommT_LAN1:
			sprintf(gAvProtoHandle->MacAddr, "eth1@%s", _NetCommProfile.Mac);
			break;
		case NetCommT_Wireless:
			sprintf(gAvProtoHandle->MacAddr, "wlan0@%s", _NetCommProfile.Mac);
			break;
		case NetCommT_Wireless1:
			sprintf(gAvProtoHandle->MacAddr, "wlan1@%s", _NetCommProfile.Mac);
			break;
		default:
			break;
		}
	}
	sprintf(_Connect.MacAddr, "%s", gAvProtoHandle->MacAddr);
	return _MsgConnect2Net(data, gAvProtoHandle->MsgHeadSend, &_Connect);
}



int _PushDevLogIteam(int sock, LogItem* iteam) {
	return 0;
}

int _PushDevLogIteamList(int sock) {
	return 0;
}

int _GetServiceIpAddr(char* host, char* Ipaddr) {
	char ServerIp[5][32];
	memset(ServerIp, 0x00, sizeof(ServerIp));
	int size = 5;
	int _ret = NetGetIpAddrByHost(host, ServerIp, &size);
	if (_ret == 0 && size > 0) {
		sprintf(Ipaddr, "%s", ServerIp[0]);
		return 0;
	}
	else {
		return -1;
	}
}

int _Connect()
{
	char MsgBuffer[128];
	int Length = 0;
	int ServiceSocket = 0;
	char ServerIpaddr[64] = { 0 };
	if (0 == gAvProtoHandle->bNetConnected) {
		ServiceSocket = 0;
		int _ret = _GetServiceIpAddr("wja.ukyun.top", ServerIpaddr);
		if (_ret != 0 || strlen(ServerIpaddr) <= 0) {
			return -1;
		}
		_ret = NetConnect(ServerIpaddr, 1532, 5000);
		if (_ret > 0) {
			ServiceSocket = _ret;
			gAvProtoHandle->bNetConnected = 1;
			Length = _RequestConnect(MsgBuffer);
			if (Length > 0) {
				_ret = send(ServiceSocket, MsgBuffer, Length, 0);
				if (_ret != Length) {
					_logDebugErr("send faield _ret = %d\n", _ret);
				}
			}
			return ServiceSocket;
		}
		else {
			return -1;
		}
	}
	else {
		return ServiceSocket;
	}
}
int _SystemVersion(int sock) {
	unsigned int cmd = _ukProtoGet | __MsgSystemVersion;
	C_SystemVersion _SystemVer = { 0 };
	if (NULL != gAvProtoHandle->onEvent) {
		gAvProtoHandle->onEvent(cmd, &_SystemVer);
	}
	char buf[256];
	int _ret = _MsgSystemVersion2Net(buf, gAvProtoHandle->MsgHeadSend, &_SystemVer);
	int _retSent = send(sock, buf, _ret, 0);
	if (_retSent != _ret) {
		_logDebugErr("send _SystemVersion faield sent = %d, ret = %d\n", _retSent, _ret);
		return -1;
	}
	else {
		return 0;
	}
}
int _KeepAlive(int sock)
{
	if (time(NULL) - gAvProtoHandle->LastSendKeepAliveT < 45) {
		return 0;
	}
	C_KeepAlive  _keepAlive = { 0 };
	char buf[512] = { 0 };
	int _length = _MsgKeepAlive2Net(buf, gAvProtoHandle->MsgHeadSend, &_keepAlive);
	int _ret = send(sock, buf, _length, 0);
	if (_ret != _length) {
		_logDebugErr("send func failed ret= %d  length = %d\n", _ret, _length);
	}
	else {
		gAvProtoHandle->LastSendKeepAliveT = time(NULL);
	}
	return 0;
}

void onConnectRespond(char* data, int datalen)
{
	C_Connect _Connect;
	memset(&_Connect, 0x00, sizeof(C_Connect));
	_MsgConnect2Local(data, datalen, &_Connect);
	if (NULL != gAvProtoHandle->onEvent) {
		gAvProtoHandle->onEvent(__MsgConnect, &_Connect);
	}
	unsigned long _AppRun = 0;
	AppRunMSec(_AppRun);
	_logDebugMsg("UID[%s], UTC[%ld],UUID[%s] LoopMs[%d]\n", _Connect.UID, _Connect.TimeVal_TvSec, _Connect.UUID, _AppRun - _Connect.LoopTimeStamp);
}

void ParseProtocolOnMsg(CMsgHead* _HeadRecv, int sock)
{
	switch (Command(_HeadRecv) & _ukProtoClear)
	{
	case __MsgConnect: {
		char* data = PayLoadData(_HeadRecv, 0);
		int length = PayLoadLength(_HeadRecv, 0);
		onConnectRespond(data, length);
		_SystemVersion(sock);
	}break;

	case __MsgKeepAlive:
		break;

	default:
		break;
	}
}

typedef struct {
	int bSockPair;
	int Sock;
}SockAttr;

void _SendLogItem(int sockServer, CMsgHead* MsgHead)
{
#if defined(_B_HAVE_SOCKETPAIR)
#else
	avMutexLock(&(gAvProtoHandle->LogMutex));
#endif

	int iCntSends = 100;
	QUEUE* node = NULL;
	LogIteamQueue* nodeItem = NULL;
	char Buffer[1024];
	int len, _ret;
	while (iCntSends-- && !QUEUE_EMPTY(&(gAvProtoHandle->LogList.ptr))) {
		node = QUEUE_NEXT(&(gAvProtoHandle->LogList.ptr));
		nodeItem = QUEUE_DATA(node, LogIteamQueue, ptr);
		len = _MsgDevLogIteam2Net(Buffer, MsgHead, &(nodeItem->Iteam));
		if (0 < len) {
			_ret = send(sockServer, Buffer, len, 0);
			if (_ret == len) {
				QUEUE_REMOVE(node);
				free(node);
				gAvProtoHandle->LogListCount--;
			}
			else {
				_logDebugErr("send faield len = %d, ret = %d\n", len, _ret);
				if (_ret < 0) {
					gAvProtoHandle->bNetConnected = 0;
				}
				break;
			}
		}
		else {
			break;
		}

	}
#if defined(_B_HAVE_SOCKETPAIR)
#else
	avMutexUnLock(&(gAvProtoHandle->LogMutex));
#endif
}


void* ThreadService(void* args) {
	avThreadDeteach();
	{
		gAvProtoHandle->bNetConnected = 0;
	}
	int ServiceSocket;

	CMsgHead* _HeadRecv = gAvProtoHandle->MsgHeadRecv = CMsgHeadNew();
	CMsgHead* _HeadSend = gAvProtoHandle->MsgHeadSend = CMsgHeadNew();
#if defined(_B_HAVE_SOCKETPAIR)
	NetSockPair(&(gAvProtoHandle->SockPairIn), &(gAvProtoHandle->SockPairOut));
#endif

	Zero(_HeadRecv);
	Zero(_HeadSend);

	int MsgBufferSize = sizeof(char*) * 4096;
	char* MsgBuffer = gAvProtoHandle->MsgBuffer = (char*)malloc(MsgBufferSize);
	int _ret = 0;
	fd_set _readfds, _errorfds;
	int _sockTotal = 0;
	SockAttr _sockAttr[3];
	while (1 == gAvProtoHandle->bThreadRun) {
		_sockTotal = 0;
		_ret = _Connect();
		if (_ret == 0) {
			_sockAttr[_sockTotal].bSockPair = 0;
			_sockAttr[_sockTotal].Sock = ServiceSocket;
			_sockTotal++;
		}
		else if (_ret > 0) {
			ServiceSocket = _ret;
			_sockAttr[_sockTotal].bSockPair = 0;
			_sockAttr[_sockTotal].Sock = ServiceSocket;
			_sockTotal++;
		}
		else {
			ServiceSocket = -1;
		}
		struct timeval tv;
#if defined(_B_HAVE_SOCKETPAIR)
		if (gAvProtoHandle->SockPairIn != -1) {
			_sockAttr[_sockTotal].bSockPair = 1;
			_sockAttr[_sockTotal].Sock = gAvProtoHandle->SockPairIn;
			_sockTotal++;
		}
		tv.tv_sec = 1;
		tv.tv_usec = 0;
#else
		tv.tv_sec = 0;
		tv.tv_usec = 200 * 1000;
#endif
		if (-1 != ServiceSocket) {
			_SendLogItem(ServiceSocket, gAvProtoHandle->MsgHeadSend);
			_KeepAlive(ServiceSocket);
		}

		FD_ZERO(&_readfds);
		FD_ZERO(&_errorfds);
		int _sockMax = 0;
		int i = 0;
		for (i = 0; i < _sockTotal; i++) {
			if (_sockMax < _sockAttr[i].Sock) {
				_sockMax = _sockAttr[i].Sock;
			}
			FD_SET(_sockAttr[i].Sock, &_readfds);
			FD_SET(_sockAttr[i].Sock, &_errorfds);
		}

		_ret = select(_sockMax + 1, &_readfds, NULL, &_errorfds, &tv);
		if (_ret == 0) {
			continue;
		}
		else if (_ret > 0) {
			for (i = 0; i < _sockTotal && _ret > 0; i++) {
				if (FD_ISSET(_sockAttr[i].Sock, &_readfds)) {
					if (0 == _sockAttr[i].bSockPair) {
						do {
							_ret = recv(_sockAttr[i].Sock, MsgBuffer, MsgBufferSize, 0);
							if (_ret > 0) {
								ParseData(_HeadRecv, MsgBuffer, _ret);
								while (0 == bHaveMsg(_HeadRecv)) {
									ParseProtocolOnMsg(_HeadRecv, _sockAttr[i].Sock);
									PopOneMsg(_HeadRecv);
								}
							}
						} while (_ret == MsgBufferSize);
					}
					else {
						do {
							_ret = recv(_sockAttr[i].Sock, MsgBuffer, sizeof(LogItem), 0);
							if (_ret == sizeof(LogItem)) {
								LogIteamQueue* node = (LogIteamQueue*)malloc(sizeof(LogIteamQueue));
								QUEUE_INIT(&(node->ptr));
								QUEUE_INSERT_TAIL(&(gAvProtoHandle->LogList.ptr), &(node->ptr));
								gAvProtoHandle->LogListCount++;
							}
							else {
								break;
							}
						} while (_ret == sizeof(LogItem));
					}
					_ret--;
				}

				if (FD_ISSET(_sockAttr[i].Sock, &_errorfds)) {
					NetClose(_sockAttr[i].Sock);
					if (0 == _sockAttr[i].bSockPair) {
						gAvProtoHandle->bNetConnected = 0;
						ServiceSocket = -1;
					}
					_ret--;
				}
			}


		}
	}



	if (NULL != _HeadRecv) {
		CMsgHeadRelease(_HeadRecv);
		gAvProtoHandle->MsgHeadRecv = _HeadRecv = NULL;
	}

	if (NULL != _HeadSend) {
		CMsgHeadRelease(_HeadSend);
		gAvProtoHandle->MsgHeadSend = _HeadSend = NULL;
	}

	if (NULL != MsgBuffer) {
		free(MsgBuffer);
		gAvProtoHandle->MsgBuffer = MsgBuffer = NULL;
	}

	return NULL;
}


int  AvProtoServiceStart(char* uid, char* uuid, onServiceEvent onCallBack)
{
#if defined(WIN32)
	WORD sockVersion = MAKEWORD(2, 2);
	WSADATA wsaData;
	if (WSAStartup(sockVersion, &wsaData) != 0) {
		return 0;
	}
#endif

	if (NULL != gAvProtoHandle) {
		return -1;
	}
	else {
		gAvProtoHandle = (AvProtoHandle*)malloc(sizeof(AvProtoHandle));
		{
			gAvProtoHandle->bThreadRun = 1;
			QUEUE_INIT(&(gAvProtoHandle->LogList.ptr));
			if (NULL != uid) {
				sprintf(gAvProtoHandle->uid, "%s", uid);
			}
			else {
				sprintf(gAvProtoHandle->uid, "%s", "");
			}

			if (NULL != uuid) {
				sprintf(gAvProtoHandle->uuid, "%s", uuid);
			}
			else {
				sprintf(gAvProtoHandle->uuid, "%s", "");
			}

			memset(gAvProtoHandle->MacAddr, 0x00, sizeof(gAvProtoHandle->MacAddr));
			gAvProtoHandle->LastSendKeepAliveT = time(NULL);
			gAvProtoHandle->bNetConnected = 0;

			gAvProtoHandle->MsgHeadRecv = NULL;
			gAvProtoHandle->MsgHeadSend = NULL;
			gAvProtoHandle->MsgBuffer = NULL;
			gAvProtoHandle->onEvent = onCallBack;
			gAvProtoHandle->LogListCount = 0;
#if defined(_B_HAVE_SOCKETPAIR)
			gAvProtoHandle->SockPairIn = -1;
			gAvProtoHandle->SockPairOut = -1;
#else
			avMutexInit(&(gAvProtoHandle->LogMutex));
#endif
		}


		return avThreadCreate(ThreadService, NULL);
	}
}
int  AvProtoServiceStop()
{
	return 0;
}
void AvProtoPushDevLog(LogItem* iteam)
{
#if defined(_B_HAVE_SOCKETPAIR)
	if (-1 != gAvProtoHandle->SockPairOut) {
		int _ret = send(gAvProtoHandle->SockPairOut, iteam, sizeof(LogItem), 0);
		if (_ret < sizeof(LogItem)) {
			_logDebugErr("Error send _ret = %d, size = %d\n", _ret, sizeof(LogItem));
		}
	}
#else
	LogIteamQueue* que = (LogIteamQueue*)malloc(sizeof(LogIteamQueue));
	QUEUE_INIT(&(que->ptr));
	que->Iteam = *iteam;

	avMutexLock(&(gAvProtoHandle->LogMutex));
	if (gAvProtoHandle->LogListCount > LOG_MAX_CACHE_ITEM_NUM) {
		LogIteamQueue* nodeItem = NULL;
		QUEUE* del = QUEUE_HEAD(&(gAvProtoHandle->LogList.ptr));
		nodeItem = QUEUE_DATA(del, LogIteamQueue, ptr);
		QUEUE_REMOVE(del);
		free(nodeItem);
		gAvProtoHandle->LogListCount--;
	}
	QUEUE_INSERT_TAIL(&(gAvProtoHandle->LogList.ptr), &(que->ptr));
	gAvProtoHandle->LogListCount++;
	avMutexUnLock(&(gAvProtoHandle->LogMutex));
#endif

}
